package com.hr.异步调用;

import java.util.ArrayDeque;
import java.util.Deque;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * @program: leetcode
 * @description:
 * @author: HuRan QQ1345122890
 * @create: 2021-08-20 20:01
 */
public class BlockingQueueTest {
    @FunctionalInterface
    interface RejectPolicy<T> {
        void reject(BlockingQueue<T> queue, T task);
    }

    static class ThreadPool {
        private BlockingQueue<Runnable> taskQueue;
        private HashSet<Worker> workers = new HashSet<Worker>();
        private int coreSize;
        private long timeout;
        private TimeUnit unit;
        private RejectPolicy rejectPolicy;

        public ThreadPool(int coreSize, long timeout, TimeUnit unit, int cap, RejectPolicy rejectPolicy) {
            this.coreSize = coreSize;
            this.timeout = timeout;
            this.unit = unit;
            this.taskQueue = new BlockingQueue<>(cap);
            this.rejectPolicy = rejectPolicy;


        }

        public void execute(Runnable task) {
            synchronized (workers) {
                if (workers.size() < coreSize) {
                    Worker worker = new Worker(task);
                    workers.add(worker);
                    worker.start();
                } else {
                    // taskQueue.put(task);
                    taskQueue.tryPut(rejectPolicy, task);
                }
            }
        }

        class Worker extends Thread {
            private Runnable task;

            public Worker(Runnable task) {
                this.task = task;
            }

            @Override
            public void run() {
                while (task != null || (task = taskQueue.poll(timeout, unit)) != null) {
                    task.run();
                    task = null;
                }
                synchronized (workers) {
                    workers.remove(this);
                }
            }
        }
    }

    static class BlockingQueue<T> {
        private Deque<T> queue = new ArrayDeque<>();
        private ReentrantLock lock = new ReentrantLock();
        final Condition fullWait = lock.newCondition();
        final Condition emptyWait = lock.newCondition();
        private int cap;

        public BlockingQueue(int cap) {
            this.cap = cap;
        }

        public T poll(long timeout, TimeUnit unit) {

            lock.lock();
            try {
                long l = unit.toNanos(timeout);
                while (queue.isEmpty()) {
                    try {
                        if (l <= 0) {
                            return null;
                        }
                        //返回的是剩余的时间
                        l = emptyWait.awaitNanos(l);

                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                final T t = queue.removeFirst();
                fullWait.signalAll();
                return t;
            } finally {
                lock.unlock();
            }
        }

        public T take() {
            lock.lock();
            try {
                while (queue.isEmpty()) {
                    try {
                        emptyWait.await();
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
                final T t = queue.removeFirst();
                fullWait.signalAll();
                return t;
            } finally {
                lock.unlock();
            }
        }

        public void put(T element) {
            lock.lock();
            try {
                while (queue.size() == cap) {
                    fullWait.await();
                }
                queue.addLast(element);
                emptyWait.signalAll();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
            }

        }

        public int size() {
            lock.lock();
            try {
                return queue.size();
            } finally {
                lock.unlock();
            }
        }

        public void tryPut(RejectPolicy rejectPolicy, T task) {
            lock.lock();
            try {
                if (queue.size() == cap) {
                    rejectPolicy.reject(this, task);
                } else {
                    queue.addLast(task);
                    emptyWait.signalAll();
                }
            } finally {
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) {



    }
}